/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package tpc.benchmarks.job.selectJob;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import tpc.task.utils.JobBase;
import tpc.task.utils.JobUtils;
import unisb.cs.core.index.access.IndexAccesserInputFormat;
import unisb.cs.data.types.DataTypes;


//select the nation which name is France
public class SelectOnNation extends JobBase {
	
	public JobBase getInstance(){
		return new SelectOnNation();
	}

	public static class TextMap extends MapReduceBase implements Mapper<Text, Text, Text, Text> {

		public void map(Text key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
		throws IOException {
			output.collect(key, value);
		}
	} 

	public int run(String[] args) throws Exception {
		JobConf job = new JobConf(getConf(), SelectOnNation.class);
		
		job.setInputFormat(IndexAccesserInputFormat.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapperClass(TextMap.class);
		job.setNumReduceTasks(0);
		
		FileInputFormat.setInputPaths(job, args[0]); 				// input path
		FileOutputFormat.setOutputPath(job, new Path(args[1])); 	// output path

		job.set(IndexAccesserInputFormat.IDX_ATTR_TYPE, String.valueOf(DataTypes.INT_TYPE));
		job.set(IndexAccesserInputFormat.IDX_ATTR_SIZE, "4");
		job.set(IndexAccesserInputFormat.IDX_ATTR, "0");		
		job.set(IndexAccesserInputFormat.LOW_KEY, "2");
		job.set(IndexAccesserInputFormat.HIGH_KEY, "2");
		
		job.set("table.name", "tpc.data.tableRecord.NationRecord");

		job.setJobName("SelectOnNation2");
		JobClient.runJob(job);		
		return 0;
	}
	
	public static void main(String [] args) throws Exception{
		
//		String tableName = "tpc.data.tableRecord.NationRecord";
//		String inputPath1 = "nation/table";
//		String outputPath1 = "nation/binary";
//		String maxRecordLength = "148";
//		String numberOfReducer = "2";
//		
//		JobUtils.runBinaryConverter(tableName, 
//									inputPath1, 
//									outputPath1, 
//									maxRecordLength, 
//									numberOfReducer);

		String outputPath2 = "nation/index";
//		String indexAttributeIndex = "2";
//		String indexAttributeType = String.valueOf(DataTypes.CHAR_TYPE);
//		
//		JobUtils.runIndexBuilder(tableName, 
//								 outputPath1, 
//								 outputPath2, 
//								 indexAttributeIndex, 
//								 indexAttributeType, 
//								 numberOfReducer);
		
		String [] intputArgs = {outputPath2, "nation/selectResult"};
		JobUtils.runJob(new SelectOnNation(), intputArgs);
	}
}